
package graphpartitioner;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Partitioner;

/**
 * Graph partitioner.  In the context of graph algorithms, ensures that nodes
 * in the same partition/cluster are sent to the same reducer.  Example:
 * 		Keys:					Sent to Reducer:
 * 		cluster_a/node_1		cluster_a
 * 		cluster_a/node_3		cluster_a
 * 		cluster_b/node_2		cluster_b
 * 
 */
public class GraphPartitioner<K, V> extends Partitioner<Text, Writable> {
	
	public final String PARTITION_SEP = "/";
	
	public int getPartition(Text key, Writable value, int numReduceTasks) {
	
		String nodeId = key.toString();
		int partIndex = nodeId.indexOf(PARTITION_SEP);
		
		// If the data isn't partitioned, then use default HashPartitioner.
		if (partIndex == -1)
			return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
		else
			// Otherwise, partition on the cluster identifier.
			return ((nodeId.substring(0, partIndex).hashCode() & Integer.MAX_VALUE) 
					% numReduceTasks);
	}
}